package cn.me.alphamq.common.event;

import cn.me.alphamq.common.event.model.Event;

import java.util.*;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 事件总线
 * 有两个作用：
 * - 存储订阅各种 Event 的监听器
 * - 当有 Event 时，将 Event 发布给对应的所有监听器
 *
 * @author f
 */
public class EventBus {

    /**
     * 存储订阅各种 Event 的监听器的表
     * TODO 如果发现 list 中只有一个监听器，则去掉 list，让 class 直接映射 listener
     */
    private static final Map<Class<? extends Event>, List<Listener<? extends Event>>> LISTENER_MAP
            = new ConcurrentHashMap<>();

    /**
     * 处理 发布 Event 的线程池
     */
    private final ThreadPoolExecutor THREAD_POOL;

    public EventBus(String taskName) {
        this.THREAD_POOL = new ThreadPoolExecutor(
                10, 100, 3, TimeUnit.SECONDS, new ArrayBlockingQueue<>(1000),
                r -> new Thread(r, taskName + "-" + UUID.randomUUID())
        );
    }

    /**
     * 注册监听器到表中
     *
     * @param clazz    监听器监听的事件类型
     * @param listener 监听器
     */
    public static void register(Class<? extends Event> clazz, Listener<? extends Event> listener) {
        LISTENER_MAP.computeIfAbsent(clazz, item -> new ArrayList<>()).add(listener);
    }

    /**
     * 当有 Event 时，将 Event 发布给对应的所有监听器
     *
     * @param event 事件
     */
    public void publish(Event event) {
        List<Listener<? extends Event>> listeners = LISTENER_MAP.get(event.getClass());
        THREAD_POOL.execute(() -> {
            try {
                for (Listener listener : listeners) {
                    listener.onReceive(event);
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        });
    }

}
